package org.elasticsearch.kafka.indexer.service.impl.examples;

import java.io.IOException;
import java.util.Set;

import org.apache.commons.lang.time.DateFormatUtils;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.kafka.indexer.exception.IndexerESNotRecoverableException;
import org.elasticsearch.kafka.indexer.exception.IndexerESRecoverableException;
import org.elasticsearch.kafka.indexer.service.ElasticSearchBatchService;
import org.elasticsearch.kafka.indexer.service.IMessageHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import com.google.common.collect.Sets;

public class ServiceOrderHandlerImpl implements IMessageHandler {

//		private static final Logger logger = LoggerFactory.getLogger(ServiceOrderHandlerImpl.class);
		@Autowired
		private ElasticSearchBatchService elasticSearchBatchService = null;
		@Value("${esIndexName:my_index}")
		private String indexName;
		@Value("${esIndexType:varnish}")
		private String indexType;

		@Override
		public byte[] transformMessage(byte[] inputMessage, Long offset) throws Exception {
			// do not do any transformations for this scenario - just return the message as is
			return inputMessage;
		}

		@Override
		public void addMessageToBatch(byte[] inputMessage, Long offset) throws Exception {
			String routingValue = null; // we don't need routing for this simple scenario		
			String line = new String(inputMessage).trim();
			String[] fields = line.split("\001",-1);
//			long createTime = Long.parseLong(fields[21]);
//			String indexSuffix = DateFormatUtils.format(createTime * 1000, "-yyyy.MM.dd");
			elasticSearchBatchService.addEventToBulkRequest(parseToJson(fields), indexName, indexType, fields[0], 
					routingValue);
		}

		@Override
		public boolean postToElasticSearch() throws InterruptedException, IndexerESRecoverableException, IndexerESNotRecoverableException {
			return elasticSearchBatchService.postToElasticSearch();
		}
		
		public static final String[] fieldNames = new String[]{"service_order_id","product_type_id","fixed_product_id","is_asap","source","platform","status","rc_status","end_status","abnormal_mark","flag_order","account_id","user_id","user_phone","passenger_name","passenger_phone","corporate_id","corporate_dept_id","city","reason_id","flight_number","create_time","update_time","init_time","select_car_time","arrival_time","cancel_time","car_id","car_type_id","car_type_ids","car_type","car_brand","driver_id","driver_phone","driver_name","vehicle_number","expect_start_time","expect_end_time","start_time","end_time","confirm_time","start_position","start_address","end_position","end_address","expect_start","expect_start","expect_end","expect_end","start","start","end","end","payment","refund_status","pay_method","pay_status","first_recharge_transaction_id","first_recharge_amount","coupon_member_id","coupon_name","coupon_type","coupon_facevalue","discount","version","fee_version","balance_status","payable","total_amount","deposit","loan_in_credit","pay_amount","min_amount","origin_amount","origin_sharing_amount","sharing_amount","actual_time_length","dependable_distance","mileage","system_distance","alitongxin_secret_no_x","alitongxin_subs_id","alitongxin_status","passenger_session_id","last_operator","time_length"};
		public static final Set<Integer> GEOPOINT = Sets.newHashSet(45,47,49,51);
		public static final Set<Integer> DATE_FIELDS = Sets.newHashSet(21,22,23,24,25,26,36,37,38,39,40);
		
		public static String parseToJson(String[] fields) throws IOException{
			//使用XContentBuilder创建json数据
			XContentBuilder jsonBuild = XContentFactory.jsonBuilder();
			jsonBuild.startObject();
			for(int i=0; i< fieldNames.length; i++){
				if(GEOPOINT.contains(i)){
					jsonBuild.field(fieldNames[i], fields[i] + "," + fields[++i]);
				}
//				else if(DATE_FIELDS.contains(i) && StringUtils.isNumeric(fields[i])){
//					jsonBuild.field(fieldNames[i], Long.parseLong(fields[i]));
//				}
				else{
					jsonBuild.field(fieldNames[i],fields[i]);
				}
			}
			jsonBuild.endObject();
			return jsonBuild.string();
		}
		
		public static void main(String[] args) throws IOException {
			System.err.println(DateFormatUtils.format(System.currentTimeMillis(), "-yyyy.MM.dd"));
			String log="6353888132670824537\00117\0010\0010\00110000001\0010\0017\0010\0010\0010\001249182027778\001138000021400\0019416104\00113810167834\001张立秋\00113810167834\0010\0010\001bj\0010\001\0011479379863\0011479381402\0011479379863\0011479379863\0011479380614\0010\001367601\0012\0012,3,4,5,26,37\001舒适车型\001马自达ATENZA(进口)\001347279\00113717944996\001郑嘉\001京N52XJ8\0011479379863\0010\0011479380619\0011479381398\0011479379878\001中海九号公馆3区\001中海九号公馆3区\001公益西桥-地铁站\001北京市丰台区\00139.817126\001116.297325\00139.837042\001116.370822\00139.816598849826\001116.29183458116\00139.822507866753\001116.36783013238\0011\0011\0010\0011\0010\0010.0\0010\001\0010\0010.0\0010.0\0019\00116\0011\0011\00144.76\0010.0\0010.0\00144.76\00113.0\00144.76\00143.76\00112.0\001779\0019200\0010\0019216\001\0018240668\0010\001582d8b974bdd2a51d500002f\001\0011800\001UPDATE";
			
			System.out.println(parseToJson(log.split("\001")));
		}
	}
